/*
 * Copyright 2003-2007 (c) Live Media Pty Ltd. <argot@einet.com.au> 
 *
 * This software is licensed under the Argot Public License 
 * which may be found in the file LICENSE distributed 
 * with this software.
 *
 * More information about this license can be found at
 * http://www.einet.com.au/License
 * 
 * The Developer of this software is Live Media Pty Ltd,
 * PO Box 4591, Melbourne 3001, Australia.  The license is subject 
 * to the law of Victoria, Australia, and subject to exclusive 
 * jurisdiction of the Victorian courts.
 */

package com.colony.zone.process;

import java.util.Date;
import java.util.Stack;

import com.colony.concurrent.NodeThreadPool;
import com.colony.concurrent.Queue;
import com.colony.concurrent.ThreadPool;
import com.colony.zone.CRL;
import com.colony.zone.INode;
import com.colony.zone.IZone;
import com.colony.zone.ZoneException;

/**
 *  A Node extends the functionality of a Zone to allow processing of messages.  A node
 *  Processor must be assigned to the Node and is used to process all Messages sent to
 *  the Node.  
 * 
 *  <p>Each Message is placed on the NodeProcess classes Queue.  The NodeProcess class
 *  requires an instance of the nodeProcessor to process messages sent to it.
 */
public abstract class QueueNode
implements INode,Runnable
{
	private static final int NODE_THREAD_TIMEOUT = 10000; // 10 seconds.
	private static final int NODE_LEAD_TIME = 200; // 0.2sec
	
	private transient IZone _parent;
	private transient CRL _location;

	private transient Queue _queue;
	private transient Stack _waitStack;

	private transient ThreadPool _threadPool;
	private transient long _threadTimeout;
	private transient long _lastThreadCreated;
	private transient long _leadTime;

	/**
	 * The Constructor creates a nodeProcessor instance of a given name.
	 */
	protected QueueNode() 
	throws ZoneException 
	{
		_queue = new Queue();
		_waitStack = new Stack();
		_threadTimeout = NODE_THREAD_TIMEOUT;
		_lastThreadCreated = new Date().getTime() - NODE_THREAD_TIMEOUT;
		_leadTime = NODE_LEAD_TIME;
	}

	public IZone zone() 
	{
		return _parent;
	}

	public CRL getLocation() 
	{
		return _location;
	}

	public long threadTimeout() 
	{
		return _threadTimeout;
	}

	public void setThreadTimeout(long t) 
	{
		_threadTimeout = t;
	}

	public void setThreadLeadTime(long t) 
	{
		_leadTime = t;
	}

	public long threadLeadTime() 
	{
		return _leadTime;
	}

	public int queueCount() 
	{
		return _queue.count();
	}
	
	protected ThreadPool getThreadPool()
	{
		return _threadPool;
	}

	/**
	 *  Sends a message to this Node.  Message is placed on the 
	 *  queue and control is returned to this thread.
	 */

	public void send(Object m) 
	{
		putMessage(m);
	}
	
	/**
	 * After a node is created it can be connected into a zone via the Zone
	 * addNode method. This will call this method to setup the node.
	 */
	public void bind(CRL location, IZone parent) 
	throws ZoneException 
	{
		_threadPool = new NodeThreadPool( parent.getThreadPool() );
		_location = location;
		_parent = parent;
	}

	public IZone getParent()
	{
		return _parent;
	}
	
	/**
	 * this is the worker thread. It will try and keep getting messages from the
	 * queue. If getMessage returns it will return back to the thread pool.
	 */

	public void run() 
	{
		Object element;

		while ((element = getMessage()) != null) 
		{
			try 
			{
				process(element);
			} 
			catch (RuntimeException e) 
			{
				throw e;
			}
		}
	}

	protected abstract boolean process(Object m);

	private class StackThread 
	{
		public Thread thread;
		public boolean active;
	}

	/**
	 * Used by the NodeWorker to get a message from the queue
	 */

	private Object getMessage() 
	{
		Object element = _queue.read(false);
		if (element == null) 
		{
			Thread ct = Thread.currentThread();
			StackThread st = new StackThread();
			st.thread = ct;
			st.active = false;

			try 
			{
				synchronized (_waitStack) 
				{
					_waitStack.push(st);
				}

				synchronized (ct) 
				{
					if (!st.active)
						ct.wait(_threadTimeout);
				}

				// if we timedout we are still on the stack. Better cleanup.
				synchronized (_waitStack) 
				{
					if (_waitStack.contains(st))
						_waitStack.remove(st);
				}
			} 
			catch (Exception ex) 
			{
			}

			// we either got notified or timed out. Either way.. return null or
			// message.
			return _queue.read(false);
		} else {
			return element;
		}
	}

	private void putMessage(Object m) 
	{
		// write on the queue returns number of waiting
		// threads. If waiting threads available.. all is ok.
		_queue.write(m);

		// no threads waiting on the queue. Stage two. Check
		// waitingStack. Grab a thread from the stack and restart it.

		synchronized (_waitStack) 
		{
			if (_waitStack.size() > 0) 
			{
				StackThread st = (StackThread) _waitStack.pop();
				Thread t = st.thread;
				synchronized (t) 
				{
					st.active = true;
					t.notify();
				}
				return;
			}
		}

		// no threads on the stack. They must all be busy, or we
		// don't have any. Time to get a new one from the thread pool.
		// Before creating a thread make sure we don't create too many.
		// LeadTime is the minimum amount of time for this node between
		// creating threads.

		long now = new Date().getTime();
		if (now > _lastThreadCreated + _leadTime) 
		{
			boolean success = _threadPool.assignThread(this);
			if (success)
				_lastThreadCreated = now;
		}

		// just getting a thread is enough to activate it. The thread
		// will start working for the node.
	}


}